package org.huangrui.spark.scala.core.rdd.dep

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

/**
 *
 * @Author hr
 * @Create 2024-10-19 15:40
 */
object Spark08_Part_1 {
  def main(args: Array[String]): Unit = {
    val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
    val sc = new SparkContext(sparConf)

    val lines: RDD[String] = sc.makeRDD(Array("nba world", "cba atguigu", "wnba", "nba"), 2)
    val words: RDD[String] = lines.flatMap(_.split(" "))
    val wordToOne = words.map(word => {
      println("***********************")
      (word, 1)
    })

    //    val reduceRdd = wordToOne.reduceByKey((x, y) => x + y)
    //    val reduceRdd1 = reduceRdd.reduceByKey((x, y) => x + y)
    val reduceRdd = wordToOne.reduceByKey(new MyPartitioner(), (x, y) => x + y)
    val reduceRdd1 = reduceRdd.reduceByKey(new MyPartitioner(), (x, y) => x + y)
    reduceRdd1.collect().foreach(println)

    System.out.println("计算完毕")
    // Stage = 1 + 1 = 2// Stage = 1 + 1 = 2
    // Task  = 2 + 3 = 5// Task  = 2 + 3 = 5
    // http://localhost:4040// http://localhost:4040
    Thread.sleep(100000000L)

    sc.stop()
  }

  /**
   * 自定义分区器
   * 1. 继承Partitioner
   * 2. 重写方法
   * 3. 样例类中重写了 hashCode方法，重写equals方法，保证相同数据返回相同的分区索引
   */
  case class MyPartitioner() extends Partitioner {
    // 分区数量
    override def numPartitions: Int = 3

    // 根据数据的key值返回数据所在的分区索引（从0开始）
    override def getPartition(key: Any): Int = {
      key match {
        case "nba" => 0
        case "wnba" => 1
        case _ => 2
      }
    }

  }
}
